package com.xh.shuati.rocketmq.consumer;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.github.benmanes.caffeine.cache.Cache;
import com.xh.shuati.mapper.PostMapper;
import com.xh.shuati.model.entity.PostThumb;
import com.xh.shuati.rocketmq.event.ThumbEvent;
import com.xh.shuati.service.PostThumbService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.stream.Collectors;

/**
 * 点赞消息消费者
 */
@Component
@Slf4j
@RocketMQMessageListener(
        topic = "THUMB_TOPIC",
        consumerGroup = "thumb-consumer-group",
        consumeThreadMax = 20,
        consumeTimeout = 30000L
)
public class ThumbConsumer implements RocketMQListener<List<ThumbEvent>> {

    /**
     * 重试消息队列
     */
    private static final ConcurrentLinkedQueue eventQueue = new ConcurrentLinkedQueue();

    @Resource
    private PostThumbService postThumbService;

    @Resource
    private PostMapper postMapper;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Value("${rocketmq.consumer.max-reconsume-times}")
    private int retry_time;

    /**
     * 本地Caffeine缓存
     */
    @Resource(name ="postCache")
    private Cache<String, Object> postCache;

    /**
     * 接收消息
     * @param events
     */
    @Override
    public void onMessage(List<ThumbEvent> events) {
        log.info("消费者接收到消息,准备执行");
        //1.按postID分组处理
        Set<ThumbEvent> thumbEvents = events.stream()
                .collect(Collectors.toMap(
                        e -> e.getUserId() + "_" + e.getPostId(),
                        e -> e,
                        (existing, replacement) ->
                                existing.getDateTime().isAfter(replacement.getDateTime()) ? existing : replacement
                ))
                .values().stream()
                .collect(Collectors.toSet());

        //记录帖子点赞数变化
        Map<Long, Long> countMap = new ConcurrentHashMap<>();

        //2.处理事件
        thumbEvents.forEach(event -> {
            log.info("正在向数据库插入数据");
            try {
                PostThumb postThumb = new PostThumb();
                BeanUtils.copyProperties(event, postThumb);
                postThumb.setUpdateTime(new Date());
                if(event.getIsThumb().equals(ThumbEvent.ThumbType.THUMB)){
                    postThumb.setIsThumb(1);
                }else{
                    postThumb.setIsThumb(0);
                }

                //还需要查出逻辑删除的记录，因为点赞可以跑到取消点赞后面，如果逻辑删除后的记录查找不出来导致这条脏读点赞重新入库
                LambdaQueryWrapper<PostThumb> queryWrapper = Wrappers.<PostThumb>lambdaQuery()
                        .apply("postId = {0} AND userId={1} OR isThumb = 1 AND postId = {0} AND userId={1}",event.getPostId(),event.getUserId());
                PostThumb oldthumb = postThumbService.getOne(queryWrapper);
                //没有记录，放心入库
                if(ObjectUtils.isEmpty(oldthumb)){
                    boolean saved = postThumbService.save(postThumb);
                    if (!saved){
                        log.warn("点赞入库执行重试操作");
                        retry(event);
                    }else {
                        event.setStatus(1);
                        countMap.merge(event.getPostId(), 1L, Long::sum);
                    }
                    return;
                }

                //幂等性： 如果现有的点赞数据晚于当前点赞时间，说明是脏数据，则不处理
                if(isDateAfterLocalDateTime(oldthumb.getUpdateTime(),event.getDateTime())){
                    event.setStatus(1);
                    return;
                }

                if((event.getIsThumb() == ThumbEvent.ThumbType.THUMB && oldthumb.getIsThumb() == 1) ||
                        (event.getIsThumb() == ThumbEvent.ThumbType.CANCEL_THUMB && oldthumb.getIsThumb() == 0)){
                    //状态错误 -> 没有点赞改成点赞 或 点赞改成没有点赞
                    boolean added = postThumbService.update(postThumb, queryWrapper);
                    if (!added){
                        log.warn("点赞入库执行重试操作");
                        retry(event);
                    }else {
                        event.setStatus(1);
                        countMap.merge(event.getPostId(), 1L, Long::sum);
                    }
                } else {
                    //否则状态正确 -> 修改更新时间
                    boolean updated = postThumbService.update(postThumb, queryWrapper);
                    if (!updated){
                        log.warn("更新点赞入库执行重试操作");
                        retry(event);
                    }else {
                        event.setStatus(1);
                    }
                }
            } catch (BeansException e) {
                retry(event);
            }
        });

        //最后更新博客点赞数
        batchUpdateBlogs(countMap);
    }

    /**
     * 批量更新帖子的点赞数
     * @param countMap
     */
    @Transactional(rollbackFor = Exception.class)
    public void batchUpdateBlogs(Map<Long, Long> countMap) {
        if (!countMap.isEmpty()) {
            postMapper.batchUpdateThumbCount(countMap);
        }
        //把修改的帖子从缓存中删除
        //todo 广播到帖子缓存所在节点进行缓存删除,或者canal监听进行缓存删除
        countMap.keySet().forEach(postId -> {
            postCache.invalidate(postId);
            log.info("更新帖子点赞数成功,帖子id:{},点赞数:{}", postId, countMap.get(postId));
        });
    }

    /**
     * 重试机制
     */
    public void retry(ThumbEvent event){
        event.setRetryCount(event.getRetryCount()+1);
        if(event.getRetryCount() >= retry_time){
            sendToDeadLetterQueue(event);
        }else{
            eventQueue.offer(event);
        }
    }

    /**
     * 定期重试机制
     */
    @Scheduled(fixedDelayString = "${rocketmq.consumer.max-reconsume-times}")
    public void retrySchedule(){
        if(eventQueue.isEmpty()) return;
        ConcurrentLinkedQueue<ThumbEvent> queueRetry;
        synchronized (eventQueue){
            queueRetry = new ConcurrentLinkedQueue<>(eventQueue);
            eventQueue.clear();
        }
        //记录帖子点赞数变化
        Map<Long, Long> countMap = new ConcurrentHashMap<>();

        //2.处理事件
        queueRetry.forEach(event -> {
            log.info("正在向数据库插入数据");
            try {
                PostThumb postThumb = new PostThumb();
                BeanUtils.copyProperties(event, postThumb);
                postThumb.setUpdateTime(new Date());
                if(event.getIsThumb().equals(ThumbEvent.ThumbType.THUMB)){
                    postThumb.setIsThumb(1);
                }else{
                    postThumb.setIsThumb(0);
                }

                //还需要查出逻辑删除的记录，因为点赞可以跑到取消点赞后面，如果逻辑删除后的记录查找不出来导致这条脏读点赞重新入库
                LambdaQueryWrapper<PostThumb> queryWrapper = Wrappers.lambdaQuery(PostThumb.class)
                        .apply("poseId = {0} AND userId={1} OR isThumb = 1 AND poseId = {0} AND userId={1}",event.getPostId(),event.getUserId());
                PostThumb oldthumb = postThumbService.getOne(queryWrapper);
                //没有记录，放心入库
                if(ObjectUtils.isEmpty(oldthumb)){
                    boolean saved = postThumbService.save(postThumb);
                    if (!saved){
                        log.warn("点赞入库执行重试操作");
                        retry(event);
                    }else {
                        event.setStatus(1);
                        countMap.merge(event.getPostId(), 1L, Long::sum);
                    }
                }

                //幂等性： 如果现有的点赞数据晚于当前点赞时间，说明是脏数据，则不处理
                if(isDateAfterLocalDateTime(oldthumb.getUpdateTime(),event.getDateTime())){
                    event.setStatus(1);
                    return;
                }

                if((event.getIsThumb() == ThumbEvent.ThumbType.THUMB && oldthumb.getIsThumb() == 1) ||
                        (event.getIsThumb() == ThumbEvent.ThumbType.CANCEL_THUMB && oldthumb.getIsThumb() == 0)){
                    //状态错误 -> 没有点赞改成点赞 或 点赞改成没有点赞
                    boolean added = postThumbService.update(postThumb, queryWrapper);
                    if (!added){
                        log.warn("点赞入库执行重试操作");
                        retry(event);
                    }else {
                        event.setStatus(1);
                        countMap.merge(event.getPostId(), 1L, Long::sum);
                    }
                } else {
                    //否则状态正确 -> 修改更新时间
                    boolean updated = postThumbService.update(postThumb, queryWrapper);
                    if (!updated){
                        log.warn("更新点赞入库执行重试操作");
                        retry(event);
                    }else {
                        event.setStatus(1);
                    }
                }
            } catch (BeansException e) {
                retry(event);
            }
        });

        //最后更新博客点赞数
        batchUpdateBlogs(countMap);
    }

    /**
     * 发送消息到死信队列
     * @param event
     */
    private void sendToDeadLetterQueue(ThumbEvent event) {
        try {
            rocketMQTemplate.syncSend("DEAD_LETTER_TOPIC",
                    MessageBuilder.withPayload(event)
                            .setHeader(MessageConst.PROPERTY_ORIGIN_MESSAGE_ID, event.getId())
                            .build());
            log.warn("Sent to dead letter queue: {}", event);
        } catch (Exception e) {
            log.error("Failed to send dead letter message", e);
        }
    }

    public static boolean isDateAfterLocalDateTime(Date date, LocalDateTime localDateTime) {
        // 将 Date 转换为 LocalDateTime
        LocalDateTime dateAsLocal = date.toInstant()
                .atZone(ZoneId.systemDefault())
                .toLocalDateTime();
        return dateAsLocal.isAfter(localDateTime);
    }
}

